﻿using System;
using System.Collections.Concurrent;
using System.Text;
using System.Threading;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using Quantum.Core;

namespace Quantum.RabbitMq
{
    public class Requester : RabbitMqConnetion, IRequestBehavior
    {
        private IModel _channel;
        private string _replyQueueName;
        private EventingBasicConsumer _consumer;
        private readonly BlockingCollection<string> _respQueue = new BlockingCollection<string>();
        private IBasicProperties _props;

        private void BuildRpcChannel()
        {
            _channel = Connection.CreateModel();
            _replyQueueName = _channel.QueueDeclare().QueueName;
            _consumer = new EventingBasicConsumer(_channel);

            _props = _channel.CreateBasicProperties();
            var correlationId = Guid.NewGuid().ToString();
            _props.CorrelationId = correlationId;
            _props.ReplyTo = _replyQueueName;

            _consumer.Received += (model, ea) =>
            {
                var body = ea.Body;
                var response = Encoding.UTF8.GetString(body);
                if (ea.BasicProperties.CorrelationId == correlationId)
                {
                    _respQueue.Add(response);
                }
            };
        }

        public void Send(
            string message,
            Action<string> onSucessed,
            CancellationToken cancellationToken = new CancellationToken())
        {
            BuildRpcChannel();

            Console.WriteLine(" [x] Requesting:" + message);
            var messageBytes = Encoding.UTF8.GetBytes(message);
            _channel.BasicPublish(
                exchange: "",
                routingKey: "rpc_queue",
                basicProperties: _props,
                body: messageBytes);

            _channel.BasicConsume(
                consumer: _consumer,
                queue: _replyQueueName,
                autoAck: true);

            var result = _respQueue.Take();
            onSucessed(result);
        }

        public void Send<TIn, TOut>(
            Outgoing<TIn> message,
            Action<Incoming<TOut>> onSucessed,
            CancellationToken cancellationToken = new CancellationToken())
        {
            BuildRpcChannel();

            var jsonMsg = JsonFormatter.SerializeObject(message);
            var messageBytes = Encoding.UTF8.GetBytes(jsonMsg);
            _channel.BasicPublish(
                exchange: "",
                routingKey: "rpc_queue",
                basicProperties: _props,
                body: messageBytes);

            _channel.BasicConsume(
                consumer: _consumer,
                queue: _replyQueueName,
                autoAck: true);

            var result = _respQueue.Take();
            onSucessed(JsonFormatter.DeserializeJsonToObject<Incoming<TOut>>(result.ToString()));
        }

    }
}
